Video Thumbnail
8:18
7:20
clock icon Created with Sketch. 8 minutes

Solution: Concurrency


Alberto Miño

Hi Andreas, hope you're doing well!
I came up with this solution, which is more similar to Arjan's v1 version.

************ Start code**************
from dataclasses import dataclass
from typing import Any
from time import perf_counter

import requests
import asyncio

API_KEY = "123456"

@dataclass
class UrlTemplateClient:
template: str

def get(self, data: dict[str, Any]) -> Any:
url = self.template.format(**data)
response = requests.get(url, timeout=5)
response.raise_for_status() # Raise an exception if the request failed
return response.json()

async def a_get(self, data: dict[str, Any]) -> Any:
return await asyncio.to_thread(self.get, data)

class CityNotFoundError(Exception):
pass

async def get_capital(country: str) -> str:
client = UrlTemplateClient(template="https://restcountries.com/v3/name/{country}")
response = await client.a_get({"country": country})

# The API can return multiple matches, so we just return the capital of the first match
return response[0]["capital"][0]

async def get_forecast(city: str) -> dict[str, Any]:
client = UrlTemplateClient(
template=f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={API_KEY}"
)
response = await client.a_get({"city": city})
if "main" not in response:
raise CityNotFoundError(
f"Couldn't find weather data. Check '{city}' if it exists and is correctly spelled.\n"
)
return response

def get_temperature(full_weather_forecast: dict[str, Any]) -> float:
temperature = full_weather_forecast["main"]["temp"]
return temperature - 273.15 # convert from Kelvin to Celsius

async def async_main() -> None:
countries = ["United States of America", "Australia", "Japan", "France", "Brazil","Argentina"]
start = perf_counter()
country_capitals = await asyncio.gather(*[get_capital(country) for country in countries])

capital_wheater_forecasts = await asyncio.gather(*[get_forecast(city) for city in country_capitals])

for index, country in enumerate(countries):
print(f"The capital of {country} is {country_capitals[index]}")
print(
f"The current temperature in {country_capitals[index]} is {get_temperature(capital_wheater_forecasts[index]):.1f} °C."
)

print(f'Total elapsed time: {perf_counter() - start}')

if __name__ == "__main__":
asyncio.run(async_main())

************ End code**************
I've really enjoyed this challenge.
I still have to get used to using zip.

Have a nice week end!

REPLY
Andreas [ArjanCodes Team]

Hi! I'm doing well, hope you are well!

Nice solution! I think this is a nice improvement! I have some smaller remarks to make this even better:

* For a_get method, try using more detailed naming and not shorthands. Usually, it is not beneficial to add a label that says that the function is async, it does not add anything to what the method does, only how it is executed.
* get_capital will not always return a string because it does an API call, and it might not return a country. I would suggest that you assert or check with a guard clause if something is returned. If not, either raise an exception or None
* For the get method in the dataclass, this one can have type annotation dict[str, Any] since the .json() methods returns dict[str, Any]

REPLY
Alberto Miño

Excellent recommendations! I've made the following changes in the portion of code you mentioned in your remarks:

@dataclass
class UrlTemplateClient:
template: str

def get_http(self, data: dict[str, Any]) -> dict[str, Any]:
url = self.template.format(**data)
response = requests.get(url, timeout=5)
response.raise_for_status() # Raise an exception if the request failed
return response.json()

async def get(self, data: dict[str, Any]) -> Any:
return await asyncio.to_thread(self.get_http, data)

class RestCountryApiError(Exception):
pass
class CityNotFoundError(Exception):
pass

async def get_capital(country: str) -> str:
client = UrlTemplateClient(template="https://restcountries.com/v3/name/{country}")
try:
response = await client.get({"country": country})
if response and isinstance(response, list) and "capital" in response[0]:
# The API can return multiple matches, so we just return the capital of the first match
return response[0]["capital"][0]
else:
raise RestCountryApiError(f"Invalid missing response format or missing 'capital' data")
except Exception as error:
raise RestCountryApiError(f"There was a error calling restcountries.com API: {error}")

Thanks Andreas!

REPLY
Andreas [ArjanCodes Team]

Great improvement, Alberto!

REPLY
Loïc Riegel

Hi, I just want to say that when usage of concurrency is limited and simple, a nice solution is to use the ThreadPoolExecuter or ProcessPoolExecutor objects for the concurrent.futures module (standard library). The API is very nice and it can be used as a contect manager.

I think that if you need to do the same operation in parallel (do some image processing on a collection of images, do many API calls in parallel etc...), than this is a good solution. Also, no need for async code all the way, which avoids some complexification of the code.

Asyncio is great when the program is more complex and the concurrent operations are all over the place and different

REPLY
Andreas [ArjanCodes Team]

Interesting! I have never used ThreadPoolExecuter or ProcessPoolExecutor. I will do a smaller project to try it out!

Asyncio also is widely known, which is always good (It is only a plus and not an argument in my opinion :) )

REPLY
Juan José Expósito González

For me, it is still hard to know whether or not I will be able to implement a concurrent solution in my own projects (I have one that would terriby benefit from this)...but still don't get the whole picture.

But this exercise has put a little bit more light on my understanding of this topic. This is something I definitely need to work on harder.

Thanks Arjan! Great challenge.

REPLY
Arjan Egges

Hi Juan José, concurrent programming is quite a hard topic to wrap your mind around. The best way I found to learn is by practicing with it a lot. In your own project, you could experiment with switching to concurrent programming by making the methods/functions asynchronous. You can then experiment with asyncio.gather to group a few of these concurrent functions and see what the effect is on performance. It's a step-by-step process :).

REPLY
Juan José Expósito González

Thanks for the tip! I am currently trying to switch it to concurrent. It is a small utility that analyzes some backtests. I have a function with a for loop. Each pass does all the analysis. I think I need to change the design to one function that does the analysis for just one backtest and make it asynchronous. I suspect I will have to use asyncio.gather to group all the analyzing together...
Don't know if it will work...but for sure I will get more insight into this topic.

REPLY
Nader Bazyari

I feel like this challenge single handedly took my understanding to the next level! I also love the great comments people post and your awesome replies Arjan. This whole 30 day challenge is Brilliant

REPLY
Javier Ruiz Nebot

Similar answer here. But I have kept the UrlTemplateClient code much similar to the "before" code, and I have created a decorator to make the function async.

def asynchronizer(func: Callable[..., Any]) -> Callable[..., Any]:
····async def inner(*args: Any, **kwargs: Any):
········return await asyncio.to_thread(func, *args, **kwargs)

····return inner

@asynchronizer
def get(self, data: dict[str, Any]) -> Any:
····url = self.template.format(**data)
····response = requests.get(url, timeout=5)
····response.raise_for_status() # Raise an exception if the request failed
····return response.json()

REPLY
Arjan Egges

That's a great use for a decorator. Thanks for sharing this!

REPLY
Show More